Amazon Connectの通話データをAmazon Bedrockで分析し、コールリーズンを自動分類・集計する仕組みの構築してみた

import boto3
import csv
import json
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
from collections import Counter
import io

MODEL_ID = 'anthropic.claude-3-5-sonnet-20240620-v1:0'
s3 = boto3.client('s3')
TRANSCRIPTION_BUCKET = 'amazon-connect-xxxxxxxx'
OUTPUT_BUCKET = 'xxxxx'

CATEGORIES = [
    "商品関連",
    "請求・支払い関連",
    "予約・キャンセル関連",
    "技術サポート関連",
    "会員情報関連",
    "その他"
]

def call_bedrock_model(prompt):
    client = boto3.client('bedrock-runtime')
    response = client.converse(
        modelId=MODEL_ID,
        messages=[{"role": "user", "content": [{"text": prompt}]}],
        inferenceConfig={"maxTokens": 1000, "temperature": 0}
    )
    return response['output']['message']['content'][0]['text']

def generate_yesterday_s3_prefix():
    return (datetime.now(ZoneInfo("UTC")) - timedelta(days=1)).strftime('Analysis/Voice/%Y/%m/%d/')

def list_files_in_s3_prefix(bucket_name, prefix):
    response = s3.list_objects_v2(Bucket=bucket_name, Prefix=prefix)
    return [content['Key'] for content in response.get('Contents', []) if content['Key'] != prefix]

def fetch_s3_file_content(bucket_name, file_key):
    print(f"S3 Key:{file_key}")
    response = s3.get_object(Bucket=bucket_name, Key=file_key)
    data = response['Body'].read().decode('utf-8')
    transcript = json.loads(data).get('Transcript', [])
    return [entry['Content'] for entry in transcript if entry.get('ParticipantId') in ['AGENT', 'CUSTOMER']]

def classify_reason_with_bedrock(transcript):
    conversation_text = " ".join(transcript)
    prompt = f"""
    以下の通話内容を、次の6つのカテゴリーのいずれかに分類してください。出力は、カテゴリー名のみを返してください。

    1. 商品関連
       - 商品の在庫状況確認
       - 商品の仕様や特徴に関する質問
       - 新商品の情報やリリース予定
       - 商品の返品・交換手続き
       - 商品の配送状況や追跡
       - 商品の不具合や破損に関する問い合わせ
       - ギフト包装や特別オプションのリクエスト

    2. 請求・支払い関連
       - 請求書の発行や確認
       - 支払い方法の変更や追加
       - 支払いの遅延や未払いに関する対応
       - 領収書の発行依頼
       - クレジットカードや銀行振込のトラブル
       - キャンペーンや割引の適用確認
       - 返金手続きや返金状況の確認

    3. 予約・キャンセル関連
       - サービスや商品の予約手続き
       - 予約内容の変更や確認
       - 予約のキャンセル手続き
       - キャンセルポリシーの確認
       - 予約に関するリマインダーや通知
       - 予約時のトラブル対応
       - 予約の優先順位や特典に関する質問

    4. 技術サポート関連
       - システムやアプリの操作方法
       - エラーメッセージや不具合の解決
       - ソフトウェアやファームウェアのアップデート
       - デバイスや機器の接続方法
       - アカウントやログインの問題
       - セキュリティやプライバシーに関する質問
       - 技術的なトラブルシューティング

    5. 会員情報関連
       - 会員登録やアカウント作成
       - 会員情報の変更（住所、電話番号、メールアドレスなど）
       - パスワードのリセットや変更
       - 会員ランクや特典の確認
       - 会員資格の更新や解約
       - ポイントやマイルの確認・利用
       - 会員限定サービスやイベントの案内

    6. その他
       - 上記のいずれにも該当しない問い合わせ内容

    通話内容: 「{conversation_text}」

    出力は、上記のカテゴリー名のいずれか1つのみを返してください。余計な説明や文脈は不要です。
    """
    return call_bedrock_model(prompt).strip()

def aggregate_call_reasons(bucket_name, file_keys):
    category_counts = Counter()
    for file_key in file_keys:
        transcript = fetch_s3_file_content(bucket_name, file_key)
        category = classify_reason_with_bedrock(transcript)
        print(f"Conversation: {json.dumps(transcript, ensure_ascii=False)}")
        print(f"Classified Category: {category}")
        category_counts[category] += 1

    total_count = sum(category_counts.values())
    results = []
    for category in CATEGORIES:
        count = category_counts.get(category, 0)
        percentage = f"{(count / total_count) * 100:.1f}%" if total_count > 0 else "0.0%"
        results.append({'カテゴリー': category, '割合': percentage, '問い合わせ件数': count})
    return results

def generate_csv_content(data):
    output = io.StringIO()
    writer = csv.writer(output)
    writer.writerow(['カテゴリー', '割合', '問い合わせ件数'])
    writer.writerows([[row['カテゴリー'], row['割合'], row['問い合わせ件数']] for row in data])
    return output.getvalue()

def upload_file_to_s3(bucket_name, file_key, file_path):
    s3.upload_file(file_path, bucket_name, file_key)

def lambda_handler(event, context):
    prefix = generate_yesterday_s3_prefix()
    file_keys = list_files_in_s3_prefix(TRANSCRIPTION_BUCKET, prefix)
    if not file_keys:
        return {'statusCode': 200, 'body': "前日のデータが見つかりませんでした。"}

    processed_data = aggregate_call_reasons(TRANSCRIPTION_BUCKET, file_keys)
    yesterday = (datetime.now(ZoneInfo("UTC")) - timedelta(days=1)).strftime('%Y-%m-%d')

    output_file_key = f'DailySummary/daily_summary_{yesterday}.csv'
    local_output_file = f'/tmp/daily_summary_{yesterday}.csv'

    csv_content = generate_csv_content(processed_data)
    with open(local_output_file, mode='w', encoding='utf-8') as file:
        file.write(csv_content)

    upload_file_to_s3(OUTPUT_BUCKET, output_file_key, local_output_file)
    return {'statusCode': 200, 'body': csv_content}

カテゴリー,割合,問い合わせ件数
商品関連,37.5%,3
請求・支払い関連,12.5%,1
予約・キャンセル関連,0.0%,0
技術サポート関連,25.0%,2
会員情報関連,12.5%,1
その他,12.5%,1